package com.shaozj.demo.kafka.producer;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

@Slf4j
@RestController("/provider")
//开启 springboot 事务，因为 kafka 的配置文件开启了 kafka 事务 不然会报错
@Transactional(rollbackFor = RuntimeException.class)
//@RequiredArgsConstructor
public class ProviderController {


    private final KafkaTemplate kafkaTemplate;

    public ProviderController(KafkaTemplate<Object, Object> kafkaTemplate, KafkaSendResultHandler kafkaSendResultHandler) {
        this.kafkaTemplate = kafkaTemplate;
        this.kafkaTemplate.setProducerListener(kafkaSendResultHandler);
    }

    @GetMapping("/sendMultiple")
    public void sendMultiple() {
        String message = "发送到Kafka的消息";
        for (int i = 0; i < 10; i ++) {
            kafkaTemplate.send("topic1", "发送到Kafka的消息" + i);
            log.info(message, i);
        }
    }
    @GetMapping("/send")
    public void send() {
        Map<String, Object> map = new HashMap<>();
        map.put("id", 1);
        map.put("name", "test");
        kafkaTemplate.send("topic1", map);
        kafkaTemplate.send("topic2", "发给topic2");

    }

    /**
     * Kafka 提供了多种构建消息的方式
     */
    public void sendDemo() throws ExecutionException, InterruptedException, TimeoutException {
        //后面的 get 代表同步发送，括号内时间可选，代表超过这个时间就会抛出超时异常，但仍会发送成功
        kafkaTemplate.send("topic1", "发给topic1").get(1, TimeUnit.MILLISECONDS);
        //使用 ProducerRecord 发送消息
        ProducerRecord<Object, Object> producerRecord = new ProducerRecord<>("topic.quick.demo", "use ProducerRecord to send message");
        kafkaTemplate.send(producerRecord);
        //使用 Message 发送消息
        Map<String, Object> map = new HashMap<>();
        map.put(KafkaHeaders.TOPIC, "topic.quick.demo");
        map.put(KafkaHeaders.PARTITION, 0);
        map.put(KafkaHeaders.KEY, 0);
        GenericMessage<Object> message = new GenericMessage<>("use Message to send message", new MessageHeaders(map));
        kafkaTemplate.send(message);


    }
}
